package com.bjsdzk.collector.one2one;

import com.bjsdzk.collector.constant.Constant;
import com.bjsdzk.collector.entity.CommandQueueEle;
import com.bjsdzk.collector.entity.Message;
import com.bjsdzk.collector.utils.ByteUtils;
import com.bjsdzk.collector.utils.CRC16;
import com.bjsdzk.collector.utils.TimeUtil;
import com.bjsdzk.collector.utils.Utils;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.core.shareddata.Counter;
import io.vertx.core.shareddata.SharedData;
import io.vertx.ext.mongo.BulkOperation;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.redis.RedisClient;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ModbusModuleServer extends AbstractVerticle {

    private static Logger logger = LoggerFactory.getLogger(ModbusModuleServer.class);

    private static String police ="police";
    private static String error = "error";

    RedisClient redis;
    MongoClient mongoClient;

    public ModbusModuleServer() {
    }

    public ModbusModuleServer(RedisClient redis, MongoClient mongoClient) {
        this.redis = redis;
        this.mongoClient = mongoClient;
    }

    Integer port;
    Integer dtuType;
    String dtuTypestr;
    JsonArray activeCmd;
    String heartPackage;
    String companyId;
    volatile boolean lineStatus=false;
    AtomicInteger conectCount = new AtomicInteger(0);
    List<String> cmdList;
    Map<String,String> deviceTypeList;
    JsonObject electricParmas = new JsonObject();

    EventBus eventBus = null;

    Map<String, List<Long>> timerMap = new HashMap<>();
    Map<String, List<Long>> timerOther = new HashMap<>();
    Map<String, Long> isRepeat = new HashMap<>();
    Map<String, Fault485> isTimeout = new HashMap<>();

    @Override
    public void start() {
        try {
            eventBus = vertx.eventBus();
            port = Integer.parseInt(config().getString("port"));
            dtuTypestr = config().getString("dtuType");
            dtuType = Integer.parseInt(config().getString("dtuType"));
            companyId = config().getString("companyId");
            heartPackage = config().getString("heartPackage");
            activeCmd = new JsonArray(config().getString("activeCmd"));
            cmdList = new JsonArray(config().getString("cmdList")).getList();
            deviceTypeList = new JsonObject(config().getString("deviceTypeList")).mapTo(Map.class);
            electricParmas = new JsonObject(config().getString("electricParmas"));


            NetServerOptions options = new NetServerOptions()/* .setReceiveBufferSize(8192) */.setTcpNoDelay(true)
                    .setSoLinger(100).setPort(port);
            NetServer server = vertx.createNetServer(options);
            SharedData sharedData = vertx.sharedData();

            ConcurrentLinkedQueue<CommandQueueEle> returnQueue = new ConcurrentLinkedQueue<>();
            PriorityBlockingQueue<Message> queue = new PriorityBlockingQueue<>();

            eventBus.consumer("singletonCmd" + port, msg -> {
                JsonObject cmd = (JsonObject) msg.body();
                queue.offer(new Message(cmd.getString("cmd"), 4, "", "test", false));
            });

            if (StringUtils.equals(dtuTypestr, Constant.DTUTYPE_6) ||StringUtils.equals(dtuTypestr, Constant.DTUTYPE_8)) {
                setTimeoutByde();

                eventBus.consumer("mod.connector.cmd." + port, obj -> {
                    JsonObject json = (JsonObject) obj.body();
                   String cmdStr= json.getString("cmd");
                    String source= json.getString("source");
                    String messageId= json.getString("messageId");
                    if (!lineStatus) {
                        CommandQueueEle ele = getEle(cmdStr);
                        ele.setMessageId(messageId);
                        ele.setSource(source);
                        ele.setCallback(true);
                        ele.setTime(System.currentTimeMillis());
                        ele.setCmd(cmdStr);
                        eventBus.send(Constant.MOD_CALLBACK, returncurrResult(JsonObject.mapFrom(ele).put("success", false)));
                    } else {
                        queue.offer(new Message(cmdStr, 4, source, messageId, true));
                    }
                });

                eventBus.consumer("readJob").handler(msg -> {
                    if (lineStatus) {
                        deviceTypeList.forEach((k, v) -> {
                            if (StringUtils.equals(Constant.DEVICETYPE_106, v) || StringUtils.equals(Constant.DEVICETYPE_116, v)) {
                                String currentCmd = Utils.add106cmd(k, true, "8000", 85, "");
                                queue.offer(new Message(currentCmd, 3));
                            }
                        });
                    }
                });
            }

            server.connectHandler(netSocket -> {
                Map<String, Long> keepLiveMap = new HashMap<>();
                keepLiveMap.put("alive", System.currentTimeMillis());

                String handlerID = netSocket.writeHandlerID();
                timerMap.put(handlerID, new ArrayList<>());
                timerOther.put(handlerID, new ArrayList<>());

                logger.info("port==>" + port + " connection  one" +" lineStatus:"+lineStatus);

                calcCount(sharedData, true);
                conectCount.incrementAndGet();

                TcpBufferHnadler tcpBufferHnadler = new TcpBufferHnadler(netSocket, sharedData, queue, keepLiveMap, returnQueue);
                netSocket.handler(tcpBufferHnadler);

                eventBus.localConsumer(Constant.MOD_SEND_SET_MODEL + port, res -> {
                    JsonObject body = (JsonObject) res.body();
                    cancelTimer(timerMap.get(handlerID));
                    activeCmd = new JsonArray(body.getString("activeCmd"));
                    cmdList = new JsonArray(body.getString("cmdList")).getList();
                    deviceTypeList = new JsonObject(body.getString("deviceTypeList")).mapTo(Map.class);
                    electricParmas = new JsonObject(body.getString("electricParmas"));
                    setTimeoutByde();
                    startTimer(queue, timerMap.get(handlerID));
                });

                startTimer(queue, timerMap.get(handlerID));
                startOtherTimer(returnQueue, timerOther.get(handlerID));
                startConsumer(netSocket, eventBus, timerOther.get(handlerID), queue, returnQueue);

                checkLine(30000, 180000, keepLiveMap, netSocket, timerOther.get(handlerID));

                netSocket.closeHandler(h -> {
                    downLine();
                    cancelTimer(timerMap.get(handlerID));
                    cancelTimer(timerOther.get(handlerID));
                    calcCount(sharedData, false);
                    queue.clear();
                });

                netSocket.exceptionHandler(e -> {
                    logger.info("-----------------------异常关闭下线 " + "==>port:" + port + "==>" + e.getMessage());
                    netSocket.close();
                });

            });

            server.listen(port, netServerAsyncResult -> {
                if (netServerAsyncResult.succeeded()) {
                    redis.hset("mod_" + port, "deploymentID", deploymentID(), result -> {
                        if (result.failed()) {
                            logger.error("deployment fail ==> port:" + port + " ==> " + result.cause().getMessage());
                        }
                    });

                } else {
                    logger.error("start error port:" + port + netServerAsyncResult.cause().getMessage());
                    vertx.undeploy(deploymentID());
                    eventBus.send(Constant.MOD_START_FAIL, config());
                }

            });
        }catch (Exception e){
            ByteArrayOutputStream stream = new ByteArrayOutputStream();
            e.printStackTrace(new PrintStream(stream));
            String exception = stream.toString();
            logger.error(port+"=exception="+exception);
        }
    }


    @Override
    public void stop() {
        logger.info("Collector client ModbusModuleServer undeployed!!!" + "==>port:" + port + "==>");
        downLine();
    }

    private void cancelTimer(List<Long> map){
        map.forEach(e -> {
            vertx.cancelTimer(e);
        });
    }

    public void setTimeoutByde(){
        deviceTypeList.forEach((k,v) ->{
            if (!isTimeout.containsKey(k)){
                isTimeout.put(k, new Fault485(-1, false));
            }
        });
    }

    private void startTimer(PriorityBlockingQueue<Message> queue,List<Long> timerMapEle){
        vertx.setTimer(10000, h ->{
            if (lineStatus){
                startInitialTimer(queue,timerMapEle);
            }else {
                startTimer(queue,timerMapEle);
            }
        });
    }
    private void startOtherTimer(ConcurrentLinkedQueue<CommandQueueEle> returnQueue,List<Long> timerMapEle){
        long periodicId = vertx.setPeriodic(1000, id -> {
            CommandQueueEle ele = returnQueue.peek();
            if (ele == null) {
                return;
            }
            if (ele.getTime() + 1000 * 60   < System.currentTimeMillis()) {
                returnQueue.poll();
                logger.info("cmd timeout not callback " + ele.getCmd());
                if (ele.isCallback()){
                    eventBus.send(Constant.MOD_CALLBACK, returncurrResult(JsonObject.mapFrom(ele).put("success", false)));
                }
            }
        });
        timerMapEle.add(periodicId);

    }


    public void  startInitialTimer(PriorityBlockingQueue<Message> queue,List<Long> timerMapEle) {
        int size = cmdList.size();
        if (size > 0) {
            AtomicInteger countCmd = new AtomicInteger(0);
            long periodicId = vertx.setPeriodic(3500, id -> {
                int i = countCmd.incrementAndGet();
                String cmd = cmdList.get(i % size);
                queue.offer(new Message(cmd, 1));
                if (i > size * 60 * 60) {
                    countCmd.set(0);
                }
            });
            timerMapEle.add(periodicId);
        }

        deviceTypeList.forEach((k,v) ->{
            if(StringUtils.equals(Constant.DEVICETYPE_117, v)){
                String cmd=Utils.add117cmd(k, true, "600C", 24);
                long periodicId117 =vertx.setPeriodic(15*60*1000, han ->{
                    queue.offer(new Message(cmd, 2));
                });
                timerMapEle.add(periodicId117);
            }
        });

        activeCmd.stream().filter(f -> JsonObject.mapFrom(f).getBoolean("isSend")).forEach((j) -> {
                    JsonObject json = JsonObject.mapFrom(j);
                    long periodicId2 = vertx.setPeriodic(json.getInteger("cmdRate") * 1000, id -> {
                        queue.offer(new Message(json.getString("cmd"), 1));
                    });
                    timerMapEle.add(periodicId2);
                }
        );

        long periodic485Id = vertx.setPeriodic(1000*60, id -> {
            if (lineStatus) {
                deviceTypeList.forEach((k, v) -> {
                    Fault485 fault485 = isTimeout.get(k);
                    long fault485Time = fault485.getTime();
                    boolean isSend = fault485.isMsgIsSend();
                    long currTime = System.currentTimeMillis();
                    JsonObject status = new JsonObject();
                    long l = 1000 * 60 * 10;
                    if (isSend) {
                        if (fault485Time + l > currTime) {
                            status.put("status", 1);
                            fault485.setMsgIsSend(false);
                            isTimeout.put(k, fault485);
                            eventBus.send(Constant.MOD_DEVICE485_FAIL, returncurrResult(k, status));
                        }
                    } else {

                        if (fault485Time == -1 || fault485Time + l < currTime) {
                            status.put("status", 0);
                            fault485.setMsgIsSend(true);
                            isTimeout.put(k, fault485);
                            eventBus.send(Constant.MOD_DEVICE485_FAIL, returncurrResult(k, status));
                        }
                    }

                });
            }
        });
        timerMapEle.add(periodic485Id);
    }

    private void checkLine(long time, long timeout, Map<String, Long> keepLiveMap, NetSocket netSocket,  List<Long> timerOther) {
        long overtimeId = vertx.setPeriodic(time, t -> {
            if (keepLiveMap.get("alive") != null
                    && (System.currentTimeMillis() - keepLiveMap.get("alive") > timeout) && !netSocket.writeQueueFull()) {
                timeOutClose(netSocket, keepLiveMap);
            }
        });
        timerOther.add(overtimeId);
    }

    public void timeOutClose(NetSocket netSocket, Map<String, Long> keepLiveMap) {
        String nowTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(System.currentTimeMillis()));
        String startTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(keepLiveMap.get("alive")));
        logger.info(netSocket.writeQueueFull() + "-HandlerID: " + netSocket.writeHandlerID() + "----alive: contection timeout: "
                + "==>port:" + port + "==>" + "-nowTime-" + nowTime + " -startTime- " + startTime);
        netSocket.close();
    }

    public void downLine() {
        int count = conectCount.get();
        logger.info("current server run total:" + count + "==>connect close downline port:" + port);
        if (count == 1) {
            lineStatus = false;
            eventBus.send(Constant.MOD_OFFLINE_STATUS,returncurrResult(new JsonObject().put("status", 0)) );
            redis.hset("mod_" + port, "offline", TimeUtil.getCurrentDatetime(), h -> {
            });
        }
        conectCount.decrementAndGet();

    }

    public void upLine(Map<String, Long> keepLiveMap ) {
        if (!lineStatus) {
            eventBus.send(Constant.MOD_ONLINE_STATUS
                    , returncurrResult(new JsonObject().put("status", 1)));
            lineStatus = true;
        }
        keepLiveMap.put("alive", System.currentTimeMillis());
    }

    public void calcCount(SharedData sharedData, boolean isAdd) {
        sharedData.getCounter("mycounter", res -> {
            if (res.succeeded()) {
                Counter counter = res.result();
                if (isAdd) {
                    counter.incrementAndGet(count -> {
                        if (count.succeeded()) {
                            logger.info("add after current total run count: " + count.result() + "---" );
                        }
                    });
                } else {
                    counter.decrementAndGet(count -> {
                        if (count.succeeded()) {
                            logger.info("decrement after current total run count: " + count.result() + "---" );
                        }
                    });
                }

            }
        });
    }

    public void saveMongo(String saveName, List<BulkOperation> bulkOperationList, JsonArray lasteDataList, String deviceHex) {
        mongoClient.bulkWrite(saveName, bulkOperationList, stringAsyncResult -> {
            if (stringAsyncResult.succeeded()) {
                eventBus.send(Constant.MOD_LATEST, returncurrResult(deviceHex, lasteDataList.toString()));
            } else {
                logger.error("monggoDB save faile： " + stringAsyncResult.cause().getMessage() + "==>port:" + port + "==>" + deviceHex);
            }
        });
    }

    private void startConsumer(NetSocket netSocket, EventBus eventBus, List<Long> timerOther, PriorityBlockingQueue<Message> queue, ConcurrentLinkedQueue<CommandQueueEle> returnQueue) {
        long periodicId = vertx.setPeriodic(1000, id -> {
            int queueSize=queue.size();
            Message peek = queue.peek();
            if (queue.peek() != null) {
                if (!lineStatus) {
                    logger.info("port:" + port + "==>device not conection, not send cmd ,queue not consumer count" + queueSize+"cmd:"+peek.getCmd());
                    if (queueSize > 15) {
                        queue.clear();
                    }
                    return;
                }
                if (netSocket.writeQueueFull() && queueSize > 0) {
                    logger.info("情况不对,赶快撤退mod" + "==>port:" + port + "==>" + netSocket.writeHandlerID() + "==>" + "队列未消费数:" + queueSize);
                    if (queueSize > 15) {
                        queue.clear();
                    }
                    return;
                }
            }
            Message receive = queue.poll();
            if (receive != null) {
                String cmdStr = receive.getCmd();
                if (cmdStr.startsWith("at-") || cmdStr.startsWith("AT-")) {
                    cmdStr = cmdStr.substring(3);

                    returnQueue.offer(new CommandQueueEle(System.currentTimeMillis(), cmdStr, true, "at-"));
                    if (returnQueue.peek() != null) {
                        logger.debug("queue message have not return" + "==>port:" + port);
                        return;
                    }
                    netSocket.write(cmdStr, result -> {
                        if (result.failed()) {
                            logger.debug(port + "cmd send fail-info:" + result.result());
                        }
                    });
                }
                logger.info(netSocket.writeHandlerID() + "==>队列个数:" + queue.size() + "==>port:" + port + "==>cmd:" + cmdStr);

                Buffer buffer = Buffer.buffer().appendBytes(Utils.hexStringToByteArray(cmdStr));
                netSocket.write(buffer, result -> {
                    if (result.failed()) {
                        logger.debug(port + "cmd send fail-info:" + result.cause().getMessage());
                    }
                });

                CommandQueueEle ele = getEle(cmdStr);

                ele.setMessageId(receive.getMessageId());
                ele.setSource(receive.getSource());
                ele.setCallback(receive.isCallback());
                ele.setTime(System.currentTimeMillis());
                ele.setCmd(cmdStr);
                returnQueue.add(ele);

            }
        });
        timerOther.add(periodicId);
    }

    public JsonObject returncurrResult(Object returnData){
        return new JsonObject().put("dtuType", dtuType).put("port", port).put("companyId", companyId).put("createTime",TimeUtil.getCurrentDatetime()).put("data", returnData);
    }

    public JsonObject returncurrResult(String deviceHex,Object returnData){
        return returncurrResult(returnData).put("deviceHex", deviceHex);
    }

    public CommandQueueEle getEle(String cmdStr){
        CommandQueueEle ele = new CommandQueueEle();
        if(cmdStr.startsWith("68")){
            if (StringUtils.equals(cmdStr.substring(6, 8),"03")) {
                ele.setRead(true);
                ele.setValue(String.format("%02x", Integer.parseInt(cmdStr.substring(12, 16), 16) * 2).toUpperCase());
            } else {
                ele.setRead(false);
                ele.setValue(cmdStr.substring(8, 12).toUpperCase());
            }
            ele.setDeviceHex(cmdStr.substring(4, 6));
        }else {
            if (StringUtils.equals(cmdStr.substring(2, 4),"03")) {
                ele.setRead(true);
                ele.setValue(String.format("%02x", Integer.parseInt(cmdStr.substring(8, 12), 16) * 2).toUpperCase());
            } else {
                ele.setRead(false);
                ele.setValue(cmdStr.substring(4, 8).toUpperCase());
            }
            ele.setDeviceHex(cmdStr.substring(0, 2));
        }
        return ele;
    }
    class TcpBufferHnadler implements Handler<Buffer> {

        NetSocket netSocket;
        SharedData sharedData;
        PriorityBlockingQueue<Message> queue;
        Map<String, Long> keepLiveMap;
        ConcurrentLinkedQueue<CommandQueueEle> returnQueue;

        public TcpBufferHnadler(NetSocket netSocket, SharedData sharedData,
                                PriorityBlockingQueue<Message> queue, Map<String, Long> keepLiveMap,
                                 ConcurrentLinkedQueue<CommandQueueEle> returnQueue) {
            this.netSocket = netSocket;
            this.sharedData = sharedData;
            this.queue = queue;
            this.keepLiveMap = keepLiveMap;
            this.returnQueue = returnQueue;
        }

        private Buffer data = Buffer.buffer();

        @Override
        public void handle(Buffer buffer) {
            byte[] bytes = data.appendBuffer(buffer).getBytes();
            logger.info("receive data ==>port:" + port + "==>" + Utils.bytes2String(bytes));
            data = Buffer.buffer();
            int lengthPosition = 0;
            int other;
            int totalLength;

            while (true) {
                String data2String = Utils.bytes2String(bytes);
                int heartLength = heartPackage.length();
                int dataLength = data2String.length();
                if (data2String.startsWith(heartPackage)) {
                    if (dataLength == heartLength) {
                        heartbeat(keepLiveMap);
                        break;
                    }
                    data2String = data2String.substring(heartLength, dataLength);
                    bytes = Arrays.copyOfRange(bytes, heartLength, bytes.length);
                }

                if (data2String.endsWith(heartPackage)) {
                    heartbeat(keepLiveMap);
                    bytes = Arrays.copyOfRange(bytes, heartLength, bytes.length);
                    data2String = data2String.substring(0, dataLength - heartLength);
                }
                if (data2String.endsWith(heartPackage) || data2String.startsWith(heartPackage)) {
                    continue;
                }

                String deviceHex = Utils.bytes2String(Arrays.copyOfRange(bytes, 0, 1));
                String readWriteFlag;

                if (StringUtils.equals(deviceHex, "68")) {
                    deviceHex = Utils.bytes2String(Arrays.copyOfRange(bytes, 2, 3));
                    readWriteFlag = Utils.bytes2String(Arrays.copyOfRange(bytes, 3, 4));

                    if (StringUtils.equals(readWriteFlag, "10")) {
                        lengthPosition = 0;
                        other = 9;
                    } else {
                        lengthPosition = 5;
                        other = 6;
                    }
                }else {
                    readWriteFlag=Utils.bytes2String(Arrays.copyOfRange(bytes, 1, 2));
                    if (StringUtils.equals("03", readWriteFlag)) {
                        lengthPosition = 3;
                        other = 5;
                    } else if (StringUtils.equals("10", readWriteFlag)) {
                        other = 8;

                    } else {
                        logger.info("data error no modbus please check");
                        break;
                    }
                }


                String deviceType = null;
                if (StringUtils.equals(dtuTypestr, Constant.DTUTYPE_8)) {
                    deviceType = "100";
                } else {
                    deviceTypeList.get(deviceHex);
                }

                if (lengthPosition != 0) {
                    lengthPosition = new BigInteger(
                            Utils.bytesToHex(Arrays.copyOfRange(bytes, lengthPosition - 1, lengthPosition)), 16)
                            .intValue();
                    totalLength = lengthPosition + other;
                } else {
                    totalLength = other;
                }

                if (bytes.length < totalLength) {
                    data = Buffer.buffer(bytes);
                    logger.info("data lenth not enough");
                    return;
                } else if (bytes.length == totalLength) {
                    data(bytes, keepLiveMap, returnQueue,deviceType,readWriteFlag,deviceHex);
                    break;
                } else {
                    if (data(Arrays.copyOfRange(bytes, 0, totalLength ), keepLiveMap, returnQueue,deviceType,readWriteFlag,deviceHex)) {
                        data = Buffer.buffer(Arrays.copyOfRange(bytes, totalLength, bytes.length));
                    } else {
                        data = Buffer.buffer();
                        break;
                    }
                }

            }
        }

        public void heartbeat(Map<String, Long> keepLiveMap) {
            logger.debug("HandlerID: " + netSocket.writeHandlerID() + "==>port:" + port + "==>" + "heartbeat"+" lineStatus:"+lineStatus);
            upLine( keepLiveMap );
        }


        public boolean data(byte[] bytes, Map<String, Long> keepLiveMap, ConcurrentLinkedQueue<CommandQueueEle> returnQueue,String deviceType,String readWriteFlag,String deviceHex) {
//			logger.debug("parse after data==>port:" + port + "==>"+Utils.bytes2String(bytes));
            int length = bytes.length;
            int deviceNum = Integer.parseInt(deviceHex, 16);
            String resultPK;
            if (StringUtils.equals(deviceType, Constant.DEVICETYPE_117)) {
                byte[] rawData = Arrays.copyOfRange(bytes, 1, length - 1);
                byte[] crcData = Arrays.copyOfRange(bytes, length - 1,
                        length);
                if (!CRC16.calcSumCheck(rawData, crcData)) {
                    logger.warn("117 CRC not pass,message lose.==>port:" + port + "==>");
                    return false;
                }

                if (StringUtils.equals("03", readWriteFlag)) {
                    resultPK = Utils.bytes2String(Arrays.copyOfRange(bytes, 4, 5));
                } else {
                    resultPK = Utils.bytes2String(Arrays.copyOfRange(bytes, 4, 6));
                }

            } else {
                int crc = CRC16.calcCrc16(Arrays.copyOfRange(bytes, 0, length - 2));
                // 如果CRC校验不通过,丢弃数据
                if (crc != Utils.toInt(Utils.reversal(Arrays.copyOfRange(bytes, length - 2, length)), 0)) {
                    logger.warn("CRC not pass,message lose.==>port:" + port + "==>");
                    return false;
                }
                if (StringUtils.equals("03", readWriteFlag)) {
                    resultPK = Utils.bytes2String(Arrays.copyOfRange(bytes, 2, 3));
                } else {
                    resultPK = Utils.bytes2String(Arrays.copyOfRange(bytes, 2, 4));
                }
            }

            isTimeout.get(deviceHex).setTime(System.currentTimeMillis());

            upLine(keepLiveMap);
            CommandQueueEle ele;

            do {
                ele = returnQueue.poll();
                if (ele == null) {
                    logger.info("queue not message  atuo send ");
                    break;
                }
                String value = ele.getValue();

                logger.debug("value: " +value+"    resultPK:"+resultPK);
                if (StringUtils.equals(deviceHex, ele.getDeviceHex())&&StringUtils.equals(value, resultPK)) {
                    break;
                } else if (StringUtils.equals(value, "at-")) {

                } else{
                    if (ele.isCallback()){
                        eventBus.send(Constant.MOD_CALLBACK, returncurrResult(JsonObject.mapFrom(ele).put("success", false)));
                    }
                    logger.info("dump message cmd: " +ele.getCmd());
                }

            } while (returnQueue.size() > 0);


            String messageid=ele.getMessageId();
            String eleCmd=ele.getCmd();
            logger.info("==>port:" + port + "==> before  send cmd:" + eleCmd + "==>deviceType:" + deviceType + "==>messageid:"+messageid+"==>returnqueue size" + returnQueue.size());
            if(ele.isCallback()){

                eventBus.send(Constant.MOD_CALLBACK, returncurrResult(JsonObject.mapFrom(ele).put("success", true).put("callCmd", Utils.bytes2String(bytes
                ))));
                return true;
            }
            if (StringUtils.equals(deviceType, Constant.DEVICETYPE_117)) {
                if(length==24){
                    callPolice_117(bytes,ele,deviceHex,length, deviceType);
                }else if (length==54){
                    parseCurrentData_117(bytes,deviceHex);
                }
            } else if (StringUtils.equals(deviceType, Constant.DEVICETYPE_116)||StringUtils.equals(deviceType, Constant.DEVICETYPE_106)) {
                parse106(bytes,ele, deviceHex);
                if(length==75){
                    callError_106(bytes,ele,deviceHex,length,deviceType);
                }else if(length==95){
                    callPolice_106(bytes,ele,deviceHex,length,deviceType);
                }else if (length==175){
                    parseCurrentData(eventBus, netSocket, redis, bytes, mongoClient,deviceType,deviceHex);
                }
            }  else if( StringUtils.equals(deviceType, Constant.DEVICETYPE_GANWEN)){
                bytes = Arrays.copyOfRange(bytes, 3, length - 2);
                ganwen(bytes, ele, returnQueue, deviceHex);
            }
            return true;
        }




        public  void ganwen(byte[] bytes,CommandQueueEle ele, ConcurrentLinkedQueue<CommandQueueEle> returnQueue,String deviceHex){
            List<BulkOperation> bulkOperationList = new ArrayList<>();
            JsonArray lasteDataList = new JsonArray();

            String channelType = ele.getCmd().substring(4, 8);

            Integer channel = 1;
            if (StringUtils.equals("0000", channelType)) {
                channel = 1;
            } else if (StringUtils.equals("0200", channelType)) {
                channel = 2;
            } else if (StringUtils.equals("0400", channelType)) {
                channel = 3;
            } else if (StringUtils.equals("0600", channelType)) {
                channel = 4;
            }

            DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
            String currentTime = dateFormat.format(new Date());
            for (int i = 0; i < (bytes.length) / 8; i++) {
                int max = ByteUtils.bytes2Int(bytes, i * 8, 2);
                int min = ByteUtils.bytes2Int(bytes, i * 8 + 2, 2);
                int avg = ByteUtils.bytes2Int(bytes, i * 8 + 4, 2);
                int position = ByteUtils.bytes2Int(bytes, i * 8 + 6, 2);
                if (max == 65535 || max == 31073) {
                    break;
                }
                JsonObject json = new JsonObject();
                json.put("companyId", companyId);
                json.put("max", new BigDecimal(max).divide(new BigDecimal(10), 1, BigDecimal.ROUND_HALF_UP).doubleValue());
                json.put("min", new BigDecimal(min).divide(new BigDecimal(10), 1, BigDecimal.ROUND_HALF_UP).doubleValue());
                json.put("avg", new BigDecimal(avg).divide(new BigDecimal(10), 1, BigDecimal.ROUND_HALF_UP).doubleValue());
                json.put("position", position);
                json.put("port", port);
                json.put("partition", i + 1);
                json.put("channel", channel);
//                json.put("createdAt", new JsonObject().put("$date", currentTime));
//				json.put("timeStamp",new JsonObject().put("$timestamp",new JsonObject().put("t", System.currentTimeMillis() / 1000).put("i", 1)));
                JsonObject lasteData = json.copy();
                json.put("createdAt", new JsonObject().put("$date", currentTime));
                lasteDataList.add(lasteData);
                bulkOperationList.add(BulkOperation.createInsert(json));
            }
            if (!bulkOperationList.isEmpty()) {
                saveMongo("RealTimeData", bulkOperationList, lasteDataList, deviceHex);
            }
        }

        public  void parse106(byte[]bytes,CommandQueueEle ele,String deviceHex){
            if (StringUtils.equals(ele.getCmd().substring(4, 8),"0106")){
                List<String> sendcmd=new ArrayList<>();

                int n = Utils.bytestoint(bytes, 3,2);
                n = n >>> 1;
                if((n&1)==1) {
                    sendcmd.add(Utils.add106cmd(deviceHex,true,"8990", 45,""));
                }
                n = n >>> 2;
                if((n&1)==1) {
                    sendcmd.add(Utils.add106cmd(deviceHex,true,"89BD", 35,""));
                }
                sendcmd.forEach(el ->{
                    queue.offer(new Message(el, 2));
                });
            }
        }

        public void parseCurrentData(EventBus eventBus,NetSocket netSocket,RedisClient redis,byte[] currentBytes,MongoClient mongoClient,String deviceType,String deviceNum) {

            String deviceId=Utils.bytes2String(Arrays.copyOfRange(currentBytes, 3, 9));
            List<BulkOperation> bulkOperationList = new ArrayList<>();
            JsonArray latestDataList = new JsonArray();
            // 第1路温度
            byte[] datas = Arrays.copyOfRange(currentBytes, 27, 29);

            JsonObject jsonObject = parseRealtimeData(deviceId, datas, 5, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null && !jsonObject.getBoolean("delete")) {
                jsonObject.remove("delete");
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            }
            // 第2路温度
            datas = Arrays.copyOfRange(currentBytes, 29, 31);
            jsonObject = parseRealtimeData(deviceId, datas, 6, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null && !jsonObject.getBoolean("delete")) {
                jsonObject.remove("delete");
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            }
            // 第3路温度
            datas = Arrays.copyOfRange(currentBytes, 31, 33);
            jsonObject = parseRealtimeData(deviceId, datas, 7, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null && !jsonObject.getBoolean("delete")) {
                jsonObject.remove("delete");
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            }
            // 第4路温度
            datas = Arrays.copyOfRange(currentBytes, 33, 35);
            jsonObject = parseRealtimeData(deviceId, datas, 8, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null && !jsonObject.getBoolean("delete")) {
                jsonObject.remove("delete");
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            }
            // A相电压
            datas = Arrays.copyOfRange(currentBytes, 35, 37);
            jsonObject = parseRealtimeData(deviceId, datas, 9, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电压
            datas = Arrays.copyOfRange(currentBytes, 37, 39);
            jsonObject = parseRealtimeData(deviceId, datas, 10, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电压
            datas = Arrays.copyOfRange(currentBytes, 39, 41);
            jsonObject = parseRealtimeData(deviceId, datas, 11, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电流
            datas = Arrays.copyOfRange(currentBytes, 41, 43);
            jsonObject = parseRealtimeData(deviceId, datas, 12, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电流
            datas = Arrays.copyOfRange(currentBytes, 43, 45);
            jsonObject = parseRealtimeData(deviceId, datas, 13, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电流
            datas = Arrays.copyOfRange(currentBytes, 45, 47);
            jsonObject = parseRealtimeData(deviceId, datas, 14, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 零线电流
            datas = Arrays.copyOfRange(currentBytes, 47, 49);
            jsonObject = parseRealtimeData(deviceId, datas, 15, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相有功功率
            datas = Arrays.copyOfRange(currentBytes, 49, 51);
            jsonObject = parseRealtimeData(deviceId, datas, 16, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相有功功率
            datas = Arrays.copyOfRange(currentBytes, 51, 53);
            jsonObject = parseRealtimeData(deviceId, datas, 17, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相有功功率
            datas = Arrays.copyOfRange(currentBytes, 53, 55);
            jsonObject = parseRealtimeData(deviceId, datas, 18, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 总有功功率
            datas = Arrays.copyOfRange(currentBytes, 55, 57);
            jsonObject = parseRealtimeData(deviceId, datas, 19, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相无功功率
            datas = Arrays.copyOfRange(currentBytes, 57, 59);
            jsonObject = parseRealtimeData(deviceId, datas, 26, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相无功功率
            datas = Arrays.copyOfRange(currentBytes, 59, 61);
            jsonObject = parseRealtimeData(deviceId, datas, 27, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相无功功率
            datas = Arrays.copyOfRange(currentBytes, 61, 63);
            jsonObject = parseRealtimeData(deviceId, datas, 28, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 总无功功率
            datas = Arrays.copyOfRange(currentBytes, 63, 65);
            jsonObject = parseRealtimeData(deviceId, datas, 29, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相功率因数
            datas = Arrays.copyOfRange(currentBytes, 65, 67);
            jsonObject = parseRealtimeData(deviceId, datas, 20, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相功率因数
            datas = Arrays.copyOfRange(currentBytes, 67, 69);
            jsonObject = parseRealtimeData(deviceId, datas, 21, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相功率因数
            datas = Arrays.copyOfRange(currentBytes, 69, 71);
            jsonObject = parseRealtimeData(deviceId, datas, 22, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 总功率因数
            datas = Arrays.copyOfRange(currentBytes, 71, 73);
            jsonObject = parseRealtimeData(deviceId, datas, 23, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 频率
            datas = Arrays.copyOfRange(currentBytes, 73, 75);
            jsonObject = parseRealtimeData(deviceId, datas, 30, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 电压不平衡度
            datas = Arrays.copyOfRange(currentBytes, 75, 77);
            jsonObject = parseRealtimeData(deviceId, datas, 24, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 电流不平衡度
            datas = Arrays.copyOfRange(currentBytes, 77, 79);
            jsonObject = parseRealtimeData(deviceId, datas, 25, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相有功电能
            datas = Arrays.copyOfRange(currentBytes, 79, 83);
            jsonObject = parseRealtimeData(deviceId, datas, 1, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相有功电能
            datas = Arrays.copyOfRange(currentBytes, 83, 87);
            jsonObject = parseRealtimeData(deviceId, datas, 2, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相有功电能
            datas = Arrays.copyOfRange(currentBytes, 87, 91);
            jsonObject = parseRealtimeData(deviceId, datas, 3, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 总有功电能
            datas = Arrays.copyOfRange(currentBytes, 91, 95);
            jsonObject = parseRealtimeData(deviceId, datas, 4, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相3次谐波电流
            datas = Arrays.copyOfRange(currentBytes, 95, 97);
            jsonObject = parseRealtimeData(deviceId, datas, 35, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相3次谐波电流
            datas = Arrays.copyOfRange(currentBytes, 97, 99);
            jsonObject = parseRealtimeData(deviceId, datas, 36, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相3次谐波电流
            datas = Arrays.copyOfRange(currentBytes, 99, 101);
            jsonObject = parseRealtimeData(deviceId, datas, 37, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电压3次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 101, 103);
            jsonObject = parseRealtimeData(deviceId, datas, 38, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电压3次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 103, 105);
            jsonObject = parseRealtimeData(deviceId, datas, 39, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电压3次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 105, 107);
            jsonObject = parseRealtimeData(deviceId, datas, 40, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电压5次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 107, 109);
            jsonObject = parseRealtimeData(deviceId, datas, 41, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电压5次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 109, 111);
            jsonObject = parseRealtimeData(deviceId, datas, 42, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电压5次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 111, 113);
            jsonObject = parseRealtimeData(deviceId, datas, 43, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电压7次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 113, 115);
            jsonObject = parseRealtimeData(deviceId, datas, 44, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电压7次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 115, 117);
            jsonObject = parseRealtimeData(deviceId, datas, 45, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电压7次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 117, 119);
            jsonObject = parseRealtimeData(deviceId, datas, 46, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电压9次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 119, 121);
            jsonObject = parseRealtimeData(deviceId, datas, 47, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电压9次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 121, 123);
            jsonObject = parseRealtimeData(deviceId, datas, 48, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电压9次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 123, 125);
            jsonObject = parseRealtimeData(deviceId, datas, 49, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电压11次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 125, 127);
            jsonObject = parseRealtimeData(deviceId, datas, 50, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电压11次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 127, 129);
            jsonObject = parseRealtimeData(deviceId, datas, 51, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电压11次谐波含量
            datas = Arrays.copyOfRange(currentBytes, 129, 131);
            jsonObject = parseRealtimeData(deviceId, datas, 52, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电压合格率
            datas = Arrays.copyOfRange(currentBytes, 131, 133);
            jsonObject = parseRealtimeData(deviceId, datas, 53, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电压合格率
            datas = Arrays.copyOfRange(currentBytes, 133, 135);
            jsonObject = parseRealtimeData(deviceId, datas, 54, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电压合格率
            datas = Arrays.copyOfRange(currentBytes, 135, 137);
            jsonObject = parseRealtimeData(deviceId, datas, 55, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电压波形失真度
            datas = Arrays.copyOfRange(currentBytes, 137, 139);
            jsonObject = parseRealtimeData(deviceId, datas, 56, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电压波形失真度
            datas = Arrays.copyOfRange(currentBytes, 139, 141);
            jsonObject = parseRealtimeData(deviceId, datas, 57, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电压波形失真度
            datas = Arrays.copyOfRange(currentBytes, 141, 143);
            jsonObject = parseRealtimeData(deviceId, datas, 58, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // A相电流波形失真度
            datas = Arrays.copyOfRange(currentBytes, 143, 145);
            jsonObject = parseRealtimeData(deviceId, datas, 59, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // B相电流波形失真度
            datas = Arrays.copyOfRange(currentBytes, 145, 147);
            jsonObject = parseRealtimeData(deviceId, datas, 60, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // C相电流波形失真度
            datas = Arrays.copyOfRange(currentBytes, 147, 149);
            jsonObject = parseRealtimeData(deviceId, datas, 61, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null)
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            // 剩余电流
            datas = Arrays.copyOfRange(currentBytes, 149, 151);
            jsonObject = parseRealtimeData(deviceId, datas, 63, mongoClient,deviceType,deviceNum,latestDataList);
            if (jsonObject != null) {
                bulkOperationList.add(BulkOperation.createInsert(jsonObject));
            }
            if (!bulkOperationList.isEmpty()) {
                saveMongo("RealTimeData", bulkOperationList, latestDataList, deviceNum);
            }
        }

        private double getUnit(int type) {
            double unit = 1;
            switch (type) {
                case 1:
                case 2:
                case 3:
                case 4:
                    unit = 0.0001d;
                    break;
                case 5:
                case 6:
                case 7:
                case 8:
                    unit = 0.1d;
                    break;
                case 9:
                case 10:
                case 11:
                case 24:
                case 25:
                case 30:
                case 38:
                case 39:
                case 40:
                case 41:
                case 42:
                case 43:
                case 44:
                case 45:
                case 46:
                case 47:
                case 48:
                case 49:
                case 50:
                case 51:
                case 52:
                    // TODO 电压合格率的位数
                case 53:
                case 54:
                case 55:
                case 56:
                case 57:
                case 58:
                case 59:
                case 60:
                case 61:
                    unit = 0.01d;
                    break;
                case 12:
                case 13:
                case 14:
                case 15:
                case 20:
                case 21:
                case 22:
                case 23:
                case 35:
                case 36:
                case 37:
                case 62:
                    unit = 0.001d;
                    break;
                case 16:
                case 17:
                case 18:
                case 19:
                case 26:
                case 27:
                case 28:
                case 29:
                    unit = 0.001d;
                    break;
            }
            return unit;
        }

        private int getCt(int type, int ct, int pt, int lct) {
            int i = 1;
            switch (type) {
                case 12:
                case 13:
                case 14:
                case 35:
                case 36:
                case 37:
                    i = ct;
                    break;
                case 15:
                    i = lct;
                    break;
                case 9:
                case 10:
                case 11:
                    i = pt;
                    break;
                case 1:
                case 2:
                case 3:
                case 4:
                case 16:
                case 17:
                case 18:
                case 19:
                case 26:
                case 27:
                case 28:
                case 29:
                    i = ct * pt;
                    break;
            }
            return i;
        }

        private int getDigit(int type) {
            int digit = 1;
            switch (type) {
                case 5:
                case 6:
                case 7:
                case 8:
                    digit = 1;
                    break;
                case 1:
                case 2:
                case 3:
                case 4:
                case 9:
                case 10:
                case 11:
                    digit = 2;
                    break;
                case 12:
                case 13:
                case 14:
                case 15:
                    digit = 3;
                    break;
                case 16:
                case 17:
                case 18:
                case 19:
                case 26:
                case 27:
                case 28:
                case 29:
                    digit = 3;
                    break;
                case 20:
                case 21:
                case 22:
                case 23:
                case 35:
                case 36:
                case 37:
                case 62:
                    digit = 3;
                    break;
                case 24:
                case 25:
                case 30:
                    digit = 2;
                    break;
                case 38:
                case 39:
                case 40:
                case 41:
                case 42:
                case 43:
                case 44:
                case 45:
                case 46:
                case 47:
                case 48:
                case 49:
                case 50:
                case 51:
                case 52:
                case 53:
                case 54:
                case 55:
                case 56:
                case 57:
                case 58:
                case 59:
                case 60:
                case 61:
                    digit = 2;
                    break;
            }
            return digit;
        }

        private String getUnitName(int type) {
            switch (type) {
                case 1:
                case 2:
                case 3:
                case 4:
                    return "kWh";
                case 5:
                case 6:
                case 7:
                case 8:
                    return "℃";
                case 9:
                case 10:
                case 11:
                case 91:
                case 101:
                case 111:
                case 92:
                case 102:
                case 112:
                    return "V";
                case 12:
                case 13:
                case 14:
                case 15:
                case 35:
                case 36:
                case 37:
                    return "A";
                case 16:
                case 17:
                case 18:
                case 19:
                    return "kW";
                case 20:
                case 21:
                case 22:
                case 23:
                    return "";
                case 24:
                case 25:
                case 38:
                case 39:
                case 40:
                case 41:
                case 42:
                case 43:
                case 44:
                case 45:
                case 46:
                case 47:
                case 48:
                case 49:
                case 50:
                case 51:
                case 52:
                case 53:
                case 54:
                case 55:
                case 56:
                case 57:
                case 58:
                case 59:
                case 60:
                case 61:
                case 62:
                    return "%";
                case 26:
                case 27:
                case 28:
                case 29:
                    return "kVar";
                case 30:
                    return "Hz";
                case 63:
                    return "mA";
                default:
                    return "";
            }
        }

        private String getName(int type) {
            switch (type) {
                case 1:
                    return "A相正向有功电能";
                case 2:
                    return "B相正向有功电能";
                case 3:
                    return "C相正向有功电能";
                case 4:
                    return "总正向有功电能";
                case 5:
                    return "第1路温度";
                case 6:
                    return "第2路温度";
                case 7:
                    return "第3路温度";
                case 8:
                    return "第4路温度";
                case 9:
                    return "A相电压";
                case 10:
                    return "B相电压";
                case 11:
                    return "C相电压";
                case 12:
                    return "A相电流";
                case 13:
                    return "B相电流";
                case 14:
                    return "C相电流";
                case 15:
                    return "零线电流";
                case 16:
                    return "A相有功功率";
                case 17:
                    return "B相有功功率";
                case 18:
                    return "C相有功功率";
                case 19:
                    return "总有功功率";
                case 20:
                    return "A相功率因数";
                case 21:
                    return "B相功率因数";
                case 22:
                    return "C相功率因数";
                case 23:
                    return "总功率因数";
                case 24:
                    return "电压不平衡度";
                case 25:
                    return "电流不平衡度";
                case 26:
                    return "A相无功功率";
                case 27:
                    return "B相无功功率";
                case 28:
                    return "C相无功功率";
                case 29:
                    return "总无功功率";
                case 30:
                    return "频率";
                case 35:
                    return "A相3次谐波电流";
                case 36:
                    return "B相3次谐波电流";
                case 37:
                    return "C相3次谐波电流";
                case 38:
                    return "A相电压3次谐波含量";
                case 39:
                    return "B相电压3次谐波含量";
                case 40:
                    return "C相电压3次谐波含量";
                case 41:
                    return "A相电压5次谐波含量";
                case 42:
                    return "B相电压5次谐波含量";
                case 43:
                    return "C相电压5次谐波含量";
                case 44:
                    return "A相电压7次谐波含量";
                case 45:
                    return "B相电压7次谐波含量";
                case 46:
                    return "C相电压7次谐波含量";
                case 47:
                    return "A相电压9次谐波含量";
                case 48:
                    return "B相电压9次谐波含量";
                case 49:
                    return "C相电压9次谐波含量";
                case 50:
                    return "A相电压11次谐波含量";
                case 51:
                    return "B相电压11次谐波含量";
                case 52:
                    return "C相电压11次谐波含量";
                case 53:
                    return "A相电压合格率";
                case 54:
                    return "B相电压合格率";
                case 55:
                    return "C相电压合格率";
                case 56:
                    return "A相电压波形失真度";
                case 57:
                    return "B相电压波形失真度";
                case 58:
                    return "C相电压波形失真度";
                case 59:
                    return "A相电流波形失真度";
                case 60:
                    return "B相电流波形失真度";
                case 61:
                    return "C相电流波形失真度";
                case 63:
                    return "剩余电流";
                default:
                    return "";
            }
        }

        private JsonObject parseRealtimeData(String deviceId, byte[] datas, int type, MongoClient mongoClient,String deviceType,String deviceNum,JsonArray latestDataList ) {
            String valueStr;
            JsonObject warn=null;
            // 转换signed int
            if (((type >= 5 && type <= 8) || (type >= 12 && type <= 14) || (type >= 16 && type <= 23) || (type >= 26 && type <= 29)) && Utils.toBinary(datas).startsWith("1")) {
                valueStr = ((Utils.toInt(Utils.backByte(datas), 0) + 1) * -1) + "";
            } else if ((type >= 53 && type <= 55) && Utils.bytes2String(datas).equalsIgnoreCase("FFFF")) {
                valueStr = "0";
            } else if (type >= 1 && type <= 4) {
                valueStr = Utils.toLong(datas, 0) + "";
            } else {
                valueStr = Utils.toInt(datas, 0) + "";
            }
            double num=1;
            if((Constant.DEVICETYPE_116.equals(deviceType))&&((type>=9&&type<20)||(type>=35&&type<=37)||(type>25&&type<=29))) {
                num=10;
            }else if((Constant.DEVICETYPE_116.equals(deviceType))&&type<=25&&type>=20){
                //num=0.1;
            }else if((Constant.DEVICETYPE_116.equals(deviceType))&&type>=1&&type<=4){
                num=100;
            }

            double originValue = Double.parseDouble(valueStr);
            Integer	ct = (electricParmas.getInteger(deviceNum+"-ct") == null) ? 1 : electricParmas.getInteger(deviceNum+"-ct");
            Integer	pt = (electricParmas.getInteger(deviceNum+"-pt") == null) ? 1 : electricParmas.getInteger(deviceNum+"-pt");
            Integer	lct = (electricParmas.getInteger(deviceNum+"-lct") == null) ? 1 : electricParmas.getInteger(deviceNum+"-lct");
            double value = originValue * getUnit(type) * getCt(type, ct, pt, lct);
            BigDecimal bd = new BigDecimal(value*num);
            value = bd.setScale(getDigit(type), BigDecimal.ROUND_HALF_UP).doubleValue();

//        logger.info(deviceId + "采集数据【" + getName(type) + "】: " + value);

            Calendar calendar = Calendar.getInstance();

            Date happDate = calendar.getTime();
            JsonObject json = new JsonObject();
            DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
            json.put("deviceId", deviceId)
                    .put("portAndaddr", port+"-"+deviceNum)
                    .put("createdAt", new JsonObject().put("$date", dateFormat.format(happDate)))
                    .put("type", type)
                    .put("port", port)
                    .put("deviceNum", deviceNum)
                    .put("unit", getUnitName(type))
                    .put("des", getName(type))
                    .put("value", value);
            switch (type) {
                case 1:
                case 2:
                case 3:
                case 4:
                    if (originValue < 0 || originValue > 4294967295l) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        return null;
                    }
                    break;
                case 5:
                case 6:
                case 7:
                case 8:
                    json.put("delete", false);
                    if ( originValue==1990|| originValue==8880) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        json.put("delete", true);
                        break;
                    }
                    if ((originValue < -100 || originValue > 1200 )) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        return null;
                    }

                    break;
                case 9:
                case 10:
                case 11:
                case 15:
                case 24:
                case 25:
                case 30:
                case 35:
                case 36:
                case 37:
                    if (originValue < 0 || originValue > 65535) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        return null;
                    }
                    break;
                case 12:
                case 13:
                case 14:
                case 16:
                case 17:
                case 18:
                case 19:
                case 26:
                case 27:
                case 28:
                case 29:
                    if (originValue < -32767 || originValue > 32767) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        return null;
                    }
                    break;
                case 20:
                case 21:
                case 22:
                case 23:
                    if (originValue < -1000 || originValue > 1000) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        return null;
                    }
                    break;
                case 38:
                case 39:
                case 40:
                case 41:
                case 42:
                case 43:
                case 44:
                case 45:
                case 46:
                case 47:
                case 48:
                case 49:
                case 50:
                case 51:
                case 52:
                    if (originValue < -10000 || originValue > 10000) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        return null;
                    }
                    break;
                case 53:
                case 54:
                case 55:
                    if (originValue < 0 || originValue > 10000) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        return null;
                    }
                    break;
                case 56:
                case 57:
                case 58:
                case 59:
                case 60:
                case 61:
                    if (originValue < 0 || originValue > 10000) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                        return null;
                    }
                    break;
                default:
                    break;
            }

            // 只有温度和电流有预警
            if ((type >= 5 && type <= 8) || (type >= 12 && type <= 15)) {
                // 超过预警阈值
                if (warn != null && warn.getFloat(type + "") != null && value > warn.getFloat(type + "")) {
                    EventBus eventBus = vertx.eventBus();
                    eventBus.send("connector.data.warn", new JsonObject().put("deviceId", deviceId).put("warnValue", value).put("warnType", type).put("warnUp", warn.getFloat(type + "")));
                }
            }

            JsonObject latestData = new JsonObject()
                    .put("deviceId", deviceId)
                    .put("unit", getUnitName(type))
                    .put("des", getName(type))
                    .put("type", type)
                    .put("value", value)
                    .put("port", port)
                    .put("deviceNum", deviceNum)
                    .put("dtuType", dtuType)
                    .put("updatedAt", new JsonObject().put("$date", dateFormat.format(happDate)));

            latestDataList.add(latestData);

            return json;
        }



        private void parseCurrentData_117(byte[] currentBytes,String deviceHex) {
             List<BulkOperation> bulkOperationList = new ArrayList<>();
             JsonArray latestDataall = new JsonArray();
             Calendar calendar = Calendar.getInstance();
             for (int i = 5, j = 0, z = 1; i < 20; i += 2, j++, z++) {
                 byte[] dataValue = Arrays.copyOfRange(currentBytes,
                         i + (16 - j * 2) + j * 4, i + (16 - j * 2) + j * 4 + 4);
                 byte[] datas = Arrays.copyOfRange(currentBytes, i, i + 2);
                 ResBean resBean = convertResBean(datas, dataValue);
                 JsonObject jsonDate = convertBytes2Value(z, resBean
                         , calendar, mongoClient, deviceHex);
                 if (jsonDate != null) {
                     bulkOperationList.add(BulkOperation.createInsert(jsonDate));
                 }
                 JsonObject latestData = convertBytescurrent2Value(z, resBean, calendar, deviceHex);
                 if(latestData!=null){
                     latestDataall.add(latestData);
                 }

             }
             if (!bulkOperationList.isEmpty()) {
                 saveMongo("RealTimeData", bulkOperationList, latestDataall, deviceHex);
             }

         }

        private ResBean convertResBean(byte[] bytes, byte[] dataValue) {
            String bin = Utils.toBinary(bytes);
            // logger.debug(bin);
            int type = Integer.parseInt(bin.substring(0, 3), 2);
            if (type == 0) {
                return null;
            }
            // logger.debug("传感器类型: " + type);
            int state = 0;
            switch (bin.substring(3, 6)) {
                case "000":
                    // 正常
                    state = 0;
                    break;
                case "001":
                    // 越限
                    state = 1;
                    break;
                case "010":
                    // 断线
                    state = 2;
                    break;
                case "011":
                    // 短路
                    state = 3;
                    break;
                case "100":
                    // 超限
                    state = 4;
                    break;
                case "101":
                    // 故障电弧
                    state = 5;
            }
//        logger.debug("报警状态: " + state);

            if (state ==2 ||state ==3) {
                return null;
            }

            int value=0;
            if(type!=1) {
                value = Utils.toInt(dataValue, 0)/10;
            }else {
                value = Utils.toInt(dataValue, 0);
            }
//        logger.debug("传感器类型: " + type+" 报警状态: " + state+"  value-----"+value+"-dataValue-"+Utils.bytes2String(dataValue));
            ResBean resBean = new ResBean(type, state, value);
            return resBean;
        }

        class ResBean {
            private int type;
            private int state;
            private int value;

            public ResBean(int type, int state, int value) {
                this.type = type;
                this.state = state;
                this.value = value;
            }

            public ResBean(int type, int state) {
                this.type = type;
                this.state = state;
            }

            public int getType() {
                return type;
            }

            public void setType(int type) {
                this.type = type;
            }

            public int getState() {
                return state;
            }

            public void setState(int state) {
                this.state = state;
            }

            public int getValue() {
                return value;
            }

            public void setValue(int value) {
                this.value = value;
            }
        }

        private JsonObject convertBytescurrent2Value(int channel, ResBean resBean, Calendar calendar, String deviceNum) {
            JsonObject latestData = new JsonObject()
                    .put("portAndaddr", port + "-" + deviceNum)
                    .put("port", port)
                    .put("deviceHex", deviceNum)
                    .put("dtuType", dtuType)
                    .put("createdAt", calendar.getTime().getTime())
                    .put("type", resBean.getType() * 100 + channel)
                    .put("channel", channel)
                    .put("unit", getUnitName2(resBean.getType()))
                    .put("value", resBean.getValue());
            return latestData;
        }
        private JsonObject convertBytes2Value(int channel, ResBean resBean, Calendar calendar, MongoClient mongoClient, String deviceNum) {

            if (resBean == null) {
                return null;
            }

            DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
            JsonObject json = new JsonObject()
                    .put("portAndaddr", port+"-"+deviceNum)
                    .put("port", port)
                    .put("deviceHex", deviceNum)
                    .put("createdAt", new JsonObject().put("$date", dateFormat.format(calendar.getTime())))
                    .put("type", resBean.getType() * 100 + channel)
                    .put("channel", channel)
                    .put("unit", getUnitName2(resBean.getType()))
                    .put("value", resBean.getValue());

            switch (resBean.getType()) {
                case 1:case 3:case 4:
                    if (resBean.getValue() < 0 || resBean.getValue() > 1000) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                    }
                    break;
                case 2:
                    if (resBean.getValue() < -10 || resBean.getValue() > 140) {
                        mongoClient.insert("ExceptionData", json, (s) -> {});
                    }
                    break;
                default:
                    break;
            }

            return json;
        }

        private String getUnitName2(int type) {
            switch (type) {
                case 1:
                    return "mA";
                case 2:
                    return "℃";
                case 3: case 4:
                    return "A";
                default:
                    return "";
            }
        }

        public  void callPolice_117(byte[]bytes,CommandQueueEle ele,String deviceHex,int length,String deviceType) {
            //68F0030318
            // 0202
            // 043A
            // 18110715292904011912230956115000900090001A
            JsonArray policeData = new JsonArray();
            JsonArray errorData = new JsonArray();
            for (int i = 5; i < length-1; i += 24) {
                int policeChannel = Utils.bytestoint(bytes, i);
                int policeType = Utils.bytestoint(bytes, i + 1);
                int policeValue = Utils.bytestoint(bytes, i + 2, 2);
                String policeTypestr = "";
                BigDecimal decimal = new BigDecimal(policeValue);
                if (policeType == 1) {
                    policeTypestr = "过流";
                    decimal= setBig(decimal, 0);
                } else if (policeType == 2) {
                    policeTypestr = "过温";
                    decimal=setBig(decimal, 1);
                } else if (policeType == 3) {
                    policeTypestr = "电弧";
                }
                float floatValue = decimal.floatValue();

                long time = getTime(bytes, i + 4);

                String policeTypestrla = policeTypestr;
                isProduceError(police,deviceHex, policeType, policeChannel, time, policeData, i, (idd) -> {
                    return new JsonObject().put("happensTime", time).put("happensType", policeType).put("happensChannel", policeChannel).put("happensValue", floatValue).put("happensName", policeTypestrla);
                });

                int errorChannel = Utils.bytestoint(bytes, i + 10);
                int errorType = Utils.bytestoint(bytes, i + 11);

                if (errorType == 1) {
                    policeTypestr = "断路";
                } else if (errorType == 2) {
                    policeTypestr = "短路";
                } else if (errorType == 3) {
                    policeTypestr = "超限";
                }

                long errortime = getTime(bytes, i + 12);
                String errorTypestrla = policeTypestr;
                boolean bolfault = isProduceError(error,deviceHex, errorType, errorChannel, errortime, errorData, i, (idd) -> {
                    return new JsonObject().put("happensTime", errortime).put("happensType", errorType).put("happensChannel", errorChannel).put("happensName", errorTypestrla);
                });
            }

            if (policeData.size() > 0) {
                eventBus.send(Constant.MOD_DEVICE_POLICE, returnResult(deviceHex, policeData, deviceType));
            }
            if (errorData.size() > 0) {
                eventBus.send(Constant.MOD_DEVICE_FAIL, returnResult(deviceHex, errorData, deviceType));
            }
        }

        private long getTime(byte[] bytes, int i) {
            int errorYear = getPoliceTime117(bytes, i);
            int errorMonth = getPoliceTime117(bytes, i+1);
            int errorDay = getPoliceTime117(bytes, i+2);
            int errorHour = getPoliceTime117(bytes, i+3);
            int errorMin = getPoliceTime117(bytes, i+4);
            int errorSec = getPoliceTime117(bytes, i+5);
            Calendar occurTime1 = Calendar.getInstance();
            occurTime1.set(2000+errorYear, errorMonth-1, errorDay, errorHour, errorMin, errorSec);
            occurTime1.set(Calendar.MILLISECOND, 0);
            return occurTime1.getTime().getTime();
        }

        private int getPoliceTime117(byte[] bytes, int i) {
            return Integer.parseInt(Utils.bytes2String(bytes, i , 1));
        }

        public  void callPolice_106(byte[]bytes,CommandQueueEle ele,String deviceHex,int length,String deviceType) {
            //02035A0013000C0014000F000F000700050000022A001300070019000E000E001E0005000001C70000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000009108
            JsonArray returnData = new JsonArray();
            for (int i = 3; i < length; i += 18) {
                long policTime = parseDate(bytes, i,2);
                int happensType = Utils.bytestoint(bytes, i + 12, 2);
                if(happensType==0){
                    break;
                }
                int happensXiang = Utils.bytestoint(bytes, i + 14, 2);
                boolean bol = isProduceError(police,deviceHex ,happensType,happensXiang, policTime, returnData, i, (idd) -> {
                    int happensValue = Utils.bytestoint(bytes, idd + 16, 2);
                    BigDecimal decimal = new BigDecimal(happensValue);

                    if (happensType ==3 && StringUtils.equals(deviceType, Constant.DEVICETYPE_106)) {
                        decimal= setBig(decimal,3);
                    } else if(happensType==4) {
                        decimal= setBig(decimal,1);
                    }else if(happensType>=5&&happensType<=8) {
                        decimal= setBig(decimal,1);
                    }else if(happensType==14) {
                        decimal= setBig(decimal,3);
                    }else {
                        decimal=setBig(decimal,2);
                    }
                    return new JsonObject().put("happensTime", policTime).put("happensType", happensType).put("happensXiang", happensXiang).put("happensValue", decimal.floatValue());
                });

                if (bol) {
                    break;
                }
            }

            logger.debug(returnData.toString());
            if (returnData.size() > 0) {
                eventBus.send(Constant.MOD_DEVICE_POLICE, returnResult(deviceHex, returnData,deviceType));
            }

        }

        public  void callError_106(byte[]bytes,CommandQueueEle ele,String deviceHex,int length,String deviceType) {
            JsonArray returnData = new JsonArray();
            for (int i = 3; i < length; i += 16) {
                long time = parseDate(bytes, i,2);
                int happensType = Utils.bytestoint(bytes, i + 12, 2);
                if(happensType==0){
                    break;
                }
                int happensChannel = Utils.bytestoint(bytes, i + 14, 2);
                boolean bol = isProduceError(error,deviceHex , happensType,happensChannel, time, returnData, i, (idd) -> {
                    return new JsonObject().put("happensTime", time).put("happensType", happensType).put("happensChannel", happensChannel);
                });
                if (bol) {
                    break;
                }
            }
            logger.debug(returnData.toString());
            if (returnData.size() > 0) {
                eventBus.send(Constant.MOD_DEVICE_FAIL, returnResult(deviceHex, returnData,deviceType));
            }

        }

        public BigDecimal setBig(BigDecimal decimal,int n){
            double pow=Math.pow(10, n);
            decimal= decimal.divide(new BigDecimal(pow));
            decimal.setScale(n);
            return decimal;
        }

        public JsonObject returnResult(String deviceHex,Object returnData,String deviceType){
            return returncurrResult(deviceHex,returnData).put("deviceType", deviceType);
        }


        public long parseDate(byte[] bytes,int i,int j){
            int policeYear = Utils.bytestoint(bytes, i,j);
            int policeMonth = Utils.bytestoint(bytes, i+j,j);
            int policeDay = Utils.bytestoint(bytes, i+j*2,j);
            int policeHour = Utils.bytestoint(bytes, i+j*3,j);
            int policeMin = Utils.bytestoint(bytes, i+j*4,j);
            int policeSec = Utils.bytestoint(bytes, i+j*5,j);
            Calendar occurTime = Calendar.getInstance();
            occurTime.set(2000+policeYear, policeMonth-1, policeDay, policeHour, policeMin, policeSec);
            occurTime.set(Calendar.MILLISECOND, 0);
            return occurTime.getTime().getTime();
        }


        public  boolean isProduceError(String flag,String deviceHex,Object happensType,Object happensChannel, long policTime, JsonArray data,int ii, Function<Integer,JsonObject> function) {
             String policepk=deviceHex+"_"+flag+"_"+policTime+"_"+happensType+happensChannel;
            Long beforeHappensTime = isRepeat.get(policepk);
            if (beforeHappensTime == null) {
                isRepeat.put(policepk, policTime);
                data.add(function.apply(ii));
            } else {
                if (policTime > beforeHappensTime) {
                    data.add(function.apply(ii));
                } else {
                    return true;
                }
            }
            return false;
        }
    }

    class Fault485{
        @Setter
        @Getter
        private long time;
        @Setter
        @Getter
        private boolean msgIsSend;

        public Fault485(long time, boolean msgIsSend) {
            this.time = time;
            this.msgIsSend = msgIsSend;
        }
    }

    public static void main(String[] args) {
        CommandQueueEle commandQueueEle = new CommandQueueEle();
        JsonObject object = new JsonObject();
        object.put("data", JsonObject.mapFrom(commandQueueEle));
        System.out.println(object.toString());
    }
}
 